package net.lesscoding.utils;

import cn.hutool.core.collection.CollUtil;
import org.apache.rocketmq.common.message.Message;

import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * @author eleven
 * @date 2024/6/1 15:03
 * @apiNote 当前只能处理小于4M的消息，如果单条消息超过4M，会被直接抛弃
 */

public class MessageListSplitter implements Iterator<List<Message>> {
    // 批次最大大小
    private final int SIZE_LIMIT = 4 * 1024 * 1024;

    private  final List<Message> list;

    private int currentIndex;

    public MessageListSplitter(List<Message> list) {
        this.list = list;
    }


    @Override
    public boolean hasNext() {
        return currentIndex < list.size();
    }

    /**
     * 有时间可以修改一下。
     * 这里如果单挑消息超过4M的话，会被抛弃掉
     * @return
     */
    @Override
    public List<Message> next() {
        int nextIndex = currentIndex;
        int totalSize = 0;
        for (; nextIndex < list.size(); nextIndex++) {
            Message message = list.get(nextIndex);
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();

            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            tmpSize += 20;
            if (tmpSize > SIZE_LIMIT) {
                if (nextIndex - currentIndex == 0) {
                    nextIndex++;
                }
                break;
            }

            if (tmpSize + totalSize > SIZE_LIMIT) {
                break;
            } else {
                totalSize += tmpSize;
            }
        }
        // [currentIndex, nextIndex)
        List<Message> messages = list.subList(currentIndex, nextIndex);
        currentIndex = nextIndex;
        return messages;
    }

    private Integer messageLength(Message message) {
        // Message 由 Topic，Body，Log ，Properties 组成 其中Log的长度是20
        int size = 20;
        size += message.getTopic().length();
        size += message.getBody().length;
        Map<String, String> properties = message.getProperties();
        if (CollUtil.isNotEmpty(properties)) {
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                size += entry.getValue().length() + entry.getKey().length();
            }
        }
        return size;
    }
}
